#include <grpcpp/grpcpp.h>
#include "ingest.grpc.pb.h"
#include "handler.h"
#include "stream_handler.h"  // NEW: Stream handler for performance
#include "query_handler.h"
#include "ilp_handler.h"
#include "sql_writer_pq.h"
#include <iostream>
#include <thread>
#include <cstring>
#include <sys/socket.h>
#include <sys/un.h>         // NEW: For Unix Domain Socket
#include <netinet/in.h>
#include <unistd.h>
#include <sstream>

// Simple HTTP server for ILP endpoint
void handle_ilp_request(int client_fd) {
    // Check for injected delay (for timeout testing)
    const char* delay_env = std::getenv("INJECT_DELAY_MS");
    if (delay_env && *delay_env) {
        int delay_ms = std::atoi(delay_env);
        if (delay_ms > 0) {
            std::cout << "[DEBUG] Injecting delay: " << delay_ms << "ms" << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
        }
    }

    char buffer[65536] = {0};
    std::string request;
    int total_read = 0;

    // Read HTTP headers first
    while (total_read < sizeof(buffer) - 1) {
        int n = read(client_fd, buffer + total_read, sizeof(buffer) - total_read - 1);
        if (n <= 0) break;
        total_read += n;
        buffer[total_read] = '\0';

        request = std::string(buffer);
        // Check if we have complete headers
        if (request.find("\r\n\r\n") != std::string::npos) {
            break;
        }
    }

    if (total_read <= 0) {
        close(client_fd);
        return;
    }

    // Check if it's a POST to /write
    if (request.find("POST /write") != 0) {
        const char* response = "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n";
        write(client_fd, response, strlen(response));
        close(client_fd);
        return;
    }

    // Find the headers end
    size_t body_pos = request.find("\r\n\r\n");
    if (body_pos == std::string::npos) {
        const char* response = "HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n";
        write(client_fd, response, strlen(response));
        close(client_fd);
        return;
    }

    // Parse Content-Length header
    size_t content_length = 0;
    size_t cl_pos = request.find("Content-Length: ");
    if (cl_pos != std::string::npos) {
        cl_pos += 16; // Skip "Content-Length: "
        size_t cl_end = request.find("\r\n", cl_pos);
        if (cl_end != std::string::npos) {
            content_length = std::stoull(request.substr(cl_pos, cl_end - cl_pos));
        }
    }

    // Get any body data we already read
    std::string body = request.substr(body_pos + 4);

    // Read remaining body data if needed
    while (body.length() < content_length) {
        char temp_buf[4096];
        int to_read = std::min((size_t)sizeof(temp_buf), content_length - body.length());
        int n = read(client_fd, temp_buf, to_read);
        if (n <= 0) break;
        body.append(temp_buf, n);
    }

    // Parse ILP data
    std::vector<ilp::ILPPoint> points;
    if (!ilp::parse_ilp_batch(body, &points)) {
        const char* response = "HTTP/1.1 400 Bad Request\r\nContent-Length: 18\r\n\r\nInvalid ILP format";
        write(client_fd, response, strlen(response));
        close(client_fd);
        return;
    }

    // Convert to WriteRequest and write to DB
    kwdb::ingest::v1::WriteRequest write_req;
    for (const auto& ilp_point : points) {
        auto* point = write_req.add_points();
        point->set_measurement(ilp_point.measurement);
        point->set_timestamp_unix_ns(ilp_point.timestamp_ns);

        for (const auto& [k, v] : ilp_point.tags) {
            auto* tag = point->add_tags();
            tag->set_key(k);
            tag->set_value(v);
        }

        for (const auto& [k, v] : ilp_point.fields) {
            auto* field = point->add_fields();
            field->set_key(k);
            field->set_f64(v);
        }
    }

    // Write to database
    std::string err;
    long long rows_written = write_points_sql(write_req, &err);

    // Send response
    if (rows_written < 0) {
        std::string resp_body = "Error: " + err;
        std::ostringstream oss;
        oss << "HTTP/1.1 500 Internal Server Error\r\n"
            << "Content-Length: " << resp_body.length() << "\r\n"
            << "\r\n"
            << resp_body;
        std::string response = oss.str();
        write(client_fd, response.c_str(), response.length());
    } else {
        std::ostringstream oss;
        oss << "HTTP/1.1 204 No Content\r\n"
            << "X-Rows-Written: " << rows_written << "\r\n"
            << "\r\n";
        std::string response = oss.str();
        write(client_fd, response.c_str(), response.length());
    }

    close(client_fd);
}

void run_http_server() {
    int server_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (server_fd == 0) {
        std::cerr << "[http] Failed to create socket" << std::endl;
        return;
    }

    int opt = 1;
    setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    struct sockaddr_in address;
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(9091);  // HTTP on port 9091

    if (bind(server_fd, (struct sockaddr*)&address, sizeof(address)) < 0) {
        std::cerr << "[http] Bind failed on port 9091" << std::endl;
        close(server_fd);
        return;
    }

    if (listen(server_fd, 10) < 0) {
        std::cerr << "[http] Listen failed" << std::endl;
        close(server_fd);
        return;
    }

    std::cout << "[http] ILP endpoint listening on 0.0.0.0:9091/write" << std::endl;

    while (true) {
        int addrlen = sizeof(address);
        int client_fd = accept(server_fd, (struct sockaddr*)&address, (socklen_t*)&addrlen);

        if (client_fd < 0) {
            continue;
        }

        // Handle each request in a new thread
        std::thread(handle_ilp_request, client_fd).detach();
    }
}

int main(int argc, char** argv) {
  std::string tcp_addr = "0.0.0.0:9090";

  // Make UDS path configurable via environment variable
  const char* uds_path_env = getenv("KWDB_UDS_PATH");
  std::string uds_path = uds_path_env ? uds_path_env : "/tmp/kwdb-ingestd.sock";
  std::string uds_addr = "unix://" + uds_path;

  std::cout << "============================================" << std::endl;
  std::cout << "  KWDB IngestD v2.0 (Performance Enhanced)" << std::endl;
  std::cout << "============================================" << std::endl;

  // Use StreamHandler instead of IngestServiceImpl for better performance
  StreamHandler stream_service;  // NEW: High-performance stream handler
  QueryServiceImpl query_service;

  grpc::ServerBuilder builder;

  // 1. TCP listening port (remote connections)
  builder.AddListeningPort(tcp_addr, grpc::InsecureServerCredentials());
  std::cout << "[TCP] Listening on " << tcp_addr << " (remote)" << std::endl;

  // 2. Unix Domain Socket (local ultra-fast connections) ✨
  // Remove old socket file if exists
  unlink(uds_path.c_str());
  builder.AddListeningPort(uds_addr, grpc::InsecureServerCredentials());
  std::cout << "[UDS] Listening on " << uds_path << " (local fast path)" << std::endl;

  // Register services
  builder.RegisterService(&stream_service);  // Using stream handler now
  builder.RegisterService(&query_service);

  // Optional: Set max message size for large batches
  builder.SetMaxReceiveMessageSize(100 * 1024 * 1024);  // 100MB
  builder.SetMaxSendMessageSize(100 * 1024 * 1024);     // 100MB

  std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
  std::cout << "[ingestd] server started, IngestService and QueryService ready" << std::endl;

  // Start HTTP server in a separate thread
  std::thread http_thread(run_http_server);

  server->Wait();
  return 0;
}